package cn.com.guage.flink.source;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import com.alibaba.fastjson.JSON;

import cn.com.guage.flink.domain.User;

/**
* 
* @version:1.0.1
* @Description: flink读取mysql数据
* @author: yangdechao
* @date: datedate 2021年11月12日 下午3:14:03
*/
public class MysqlSourceTest {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(3);
		DataStreamSource<User> source = env.addSource(new SourceFromMySQL());
		 source.map(new MapFunction<User, String>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 8594542077228235337L;

			public String map(User value) throws Exception {
				return JSON.toJSONString(value);
			}
		}).print();
		env.execute("Flink add data sourc");
	}

}
